package com.yangjiayu.utils;

import com.yangjiayu.enums.SSEMsgType;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * @Classname SSEServer
 * @Description TODO
 * @Date 2025/8/12 22:03
 * @Created by YangJiaYu
 */
@Slf4j
public class SSEServer {

    // 存放所有用户
    private static final Map<String,SseEmitter>sseClients = new ConcurrentHashMap<>();

    /**
     * 连接SSE服务
     * @param userId
     * @return
     */
    public static SseEmitter connect(String userId){

        // 设置超时事件，0L表示永不超时；默认是30秒，超时未完成就会抛出异常
        SseEmitter emitter = new SseEmitter(0L);

        // 注册回调方法
        emitter.onTimeout(timeoutCallback(userId));
        emitter.onCompletion(completionCallback(userId));
        emitter.onError(errorCallback(userId));

        sseClients.put(userId,emitter);

        log.info("SSE连接创建成功，连接的用户ID:{}",userId);

        return emitter;


    }

    public static Consumer<Throwable> errorCallback(String userId){
        return Throwable -> {
            log.info("SSE异常...");
            // 移除用户连接
            remove(userId);
        };
    }

    public static Runnable timeoutCallback(String userId){
        return () -> {
            log.info("SSE超时...");
            // 移除用户连接
            remove(userId);
        };
    }

    public static Runnable completionCallback(String userId){
        return () -> {
            log.info("SSE完成...");
            // 移除用户连接
            remove(userId);
        };
    }

    public static void remove(String userId){
        // 删除用户
        sseClients.remove(userId);
        log.info("SSE连接被移除，移除的用户ID为：{}",userId);
    }

//    public static void sendMsg(String userId,String message,SSEMsgType msgType){
//
//        if(CollectionUtils.isEmpty(sseClients)){
//            return;
//        }
//
//        if(sseClients.containsKey(userId)){
//            SseEmitter sseEmitter = sseClients.get(userId);
//            sendEmitterMessage(sseEmitter,userId,message,msgType);
//        }
//    }

    public static void sendMsg(String userId, String message, SSEMsgType msgType) {
        log.info("准备发送消息，userId: {}, message: {}, 当前在线用户: {}", userId, message, sseClients.keySet());
        if (sseClients.containsKey(userId)) {
            SseEmitter emitter = sseClients.get(userId);
            log.info("找到 emitter，准备发送消息给 userId: {}", userId);
            sendEmitterMessage(emitter, userId, message, msgType);
        } else {
            log.warn("未找到 userId 对应的 SseEmitter: {}", userId);
        }
    }

    public static void sendEmitterMessage(SseEmitter sseEmitter,
                                        String userId,
                                        String message,
                                        SSEMsgType msgType){
        try {
            SseEmitter.SseEventBuilder msgEvent = SseEmitter.event()
                .id(userId)
                .data(message)
                .name(msgType.type);
            sseEmitter.send(msgEvent);
        } catch (IOException e) {
            log.error("SSE发送消息异常...{}",e.getMessage());
            remove(userId);
        }

    }

    /**
     * SSE服务群发消息给所有客户端
     * @param message
     */
    public static void sendMsgToAllUsers(String message){
        log.info("message: {}, 当前在线用户: {}",message, sseClients.keySet());
        if (CollectionUtils.isEmpty(sseClients.values())) {
            log.info("没人呢");
        }
        sseClients.forEach((userId,SseEmitter)->{
            sendEmitterMessage(SseEmitter,userId,message,SSEMsgType.MESSAGE);
        });
//        if(sseClients.containsKey(userId)){
//            SseEmitter sseEmitter = sseClients.get(userId);
//            sendEmitterMessage(sseEmitter,userId,message,msgType);
//        }
    }


}
